package com.lg.lock;

import lombok.SneakyThrows;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 实现zk 分布式锁 第一种解决方案
 *
 * @author by Mr. Li 2020/12/10 14:59
 */
public class ZookeeperSession {
    private static CountDownLatch connectedSemaphoreCountDownLatch = new CountDownLatch(1);

    private ZooKeeper zooKeeper;
    private CountDownLatch latch;
    private long waitTime = 200;

    private String connectString = "192.168.31.201:2181,192.168.31.201:2182,192.168.31.201:2183,192.168.31.201:2184,192.168.31.201:2185";
    private int sessionTimeout = 2000;

    public ZookeeperSession() {
        try {
            this.zooKeeper = new ZooKeeper(connectString, sessionTimeout, event -> {
                try {
                    connectedSemaphoreCountDownLatch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Zookeeper session established......");
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取锁
     *
     * @param productId 锁id
     * @return
     */
    public Boolean acquireDistributedLock(Long productId) {
        String path = getPath(productId);

        try {
            createPath(path);
            return true;
        } catch (InterruptedException | KeeperException e) {
            while (true) {
                // 创建节点失败，给node注册一个监听器，监听该节点是否存在，什么时候消失
                try {
                    Stat stat = zooKeeper.exists(path, true);
                    if (stat != null) {
                        this.latch = new CountDownLatch(1);
                        this.latch.await(waitTime, TimeUnit.MILLISECONDS); // 等待 200ms，没有获取到锁，就放弃掉
                        this.latch = null;
                    }
                    createPath(path);
                    return true;
                } catch (KeeperException | InterruptedException keeperException) {
                    continue;
                }
            }
        }
    }

    /**
     * 释放锁
     *
     * @param productId
     */
    public void releaseDistributedLock(Long productId) {
        try {
            zooKeeper.delete(getPath(productId), -1); // （如果给定版本为-1，则它与任何节点的版本相匹配）
            System.out.println("release the lock for product[id=" + productId + "]......");
        } catch (InterruptedException | KeeperException e) {
            e.printStackTrace();
        }

    }

    private class ZookeeperWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("Receive watched event: " + event.getState());

            if (Event.KeeperState.SyncConnected == event.getState()) { // 连接已创建
                connectedSemaphoreCountDownLatch.countDown();
            }
            //if (this.l)
        }
    }

    private static class Singleton {
        private static ZookeeperSession instance;

        static {
            instance = new ZookeeperSession();
        }

        public static ZookeeperSession getInstance() {
            return instance;
        }
    }

    public static ZookeeperSession getInstance() {
        return Singleton.getInstance();
    }

    public static void init() {
        getInstance();
    }


    /**
     * 创建节点
     *
     * @param path
     * @throws KeeperException
     * @throws InterruptedException
     */
    private void createPath(String path) throws KeeperException, InterruptedException {
        zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    }

    private String getPath(Long productId) {
        return "/product-lock-" + productId;
    }

    public static void main(String[] args) throws IOException {
        ZookeeperSession zookeeperSession = new ZookeeperSession();
        Boolean aBoolean = zookeeperSession.acquireDistributedLock(1L);

        System.in.read();
    }
    @Test
    public void releaseDistributedLockTest(){
        ZookeeperSession zookeeperSession = new ZookeeperSession();
        zookeeperSession.releaseDistributedLock(1L);
    }
}
